package cn.ac.iscas.service.transformer.dameng;

import cn.ac.iscas.base.utils.SpringUtil;
import cn.ac.iscas.domain.source.DamengSourceColumn;
import cn.ac.iscas.service.queue.dameng.DamengTableQueueService;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * @author LJian
 * @version 1.0
 * @description: TODO
 * @date 2022/10/8 10:10
 */
public class DamengDeduplicationTransformer {
    private final static String ID = "deduplication";
    private final static String COLUMN_KEY = "column";
    private String jobId;
    private String componentId;
    private String frontTableName;
    private List<DamengSourceColumn> frontColumns;

    private String currentTableName;
    private List<DamengSourceColumn> currentColumns;

    public DamengDeduplicationTransformer(String jobId, String taskId) {
        this.jobId = jobId;
        this.componentId = taskId;
    }

    public DamengDeduplicationTransformer(String jobId, String taskId, String frontTableName, List<DamengSourceColumn> frontColumns) {
        this.jobId = jobId;
        this.componentId = taskId;
        this.frontTableName = frontTableName;
        this.frontColumns = frontColumns;
        this.currentColumns = frontColumns;
        this.currentTableName = String.format("\"%s\".\"%s_%s\"", jobId, ID, UUID.randomUUID().toString().substring(0,6));
    }


    public DamengDeduplicationTransformer() {

    }

    /*
     * STEP1.创建临时表，示例：
       CREATE GLOBAL TEMPORARY TABLE "240102SA"."TMP_DEDUPLICATION_240102TB"
        (
            "ADDRESSID" INT  NOT NULL,
            "ADDRESS1" VARCHAR(60) NOT NULL,
            "ADDRESS2" VARCHAR(60),
            "CITY" VARCHAR(30) NOT NULL,
            "POSTALCODE" VARCHAR(15) NOT NULL,
            NOT CLUSTER PRIMARY KEY("ADDRESSID","ADDRESS1","ADDRESS2")
         );
      * STEP2.去重，示例：
        INSERT INTO "240102SA"."TMP_DEDUPLICATION_240102TB" (POSTALCODE, ADDRESSID, ADDRESS1, ADDRESS2, CITY) (SELECT DISTINCT POSTALCODE,ADDRESSID,ADDRESS1,ADDRESS2,CITY  FROM PERSON.ADDRESS);

     */
    public List<String> getSql(Map<String, Object> parameters) {
        if (!parameters.containsKey(COLUMN_KEY)) {
            return null;
        }
        List<String> sqls = new ArrayList<>();
        String tmpTableName = this.currentTableName + "tmp";

        String createTempTableHeader = String.format("CREATE GLOBAL TEMPORARY TABLE %s ( ", this.currentTableName);
        StringBuffer createTempTableBuf = new StringBuffer(createTempTableHeader);
        List<String> primaryKeys = new ArrayList<>();

        List<String> columnInOrder = new ArrayList<>();

        String distinctCol = parameters.get("column").toString();
        columnInOrder.add(distinctCol);
        for (DamengSourceColumn column : this.currentColumns) {
            String colName = column.getName();
            String colType = column.getType();
            if (colType.equalsIgnoreCase("varchar") || colType.equalsIgnoreCase("char")) {
                colType = String.format(" %s ( %s ) ", column.getType(), column.getCharMaxLen());
            }
            String colNotNul = column.getNullable() ? "" : "NOT NULL";
            if (column.isPrimaryKey()) {
                primaryKeys.add(colName);
            }

            createTempTableBuf.append(String.format(" \"%s\" %s %s , ", colName, colType, colNotNul));
            if (!colName.equalsIgnoreCase(distinctCol)) {
                columnInOrder.add(colName);
            }
        }

//        String primaryKeysStr = String.format("NOT CLUSTER PRIMARY KEY(%s)", primaryKeys.stream().map(s -> "\"" + s + "\"").collect(Collectors.joining(", ")));
//        createTempTableBuf.append(primaryKeysStr).append(");");
        String tmp = "";
        if (createTempTableBuf.toString().trim().endsWith(",")) {
            tmp = createTempTableBuf.toString().trim().substring(0, createTempTableBuf.toString().trim().length() - 1);
        }
        tmp = tmp + ");";
        sqls.add(tmp);

        StringBuffer insertDistinctBuf = new StringBuffer("INSERT INTO %s ");
        String columnInOrderStr = String.join(",", columnInOrder);
        //String distinctSelect = String.format(" SELECT %s FROM %s ", " DISTINCT " + columnInOrderStr , this.frontTableName);

        String distinctSelect = String.format(" SELECT %s FROM (SELECT %s, ROW_NUMBER() OVER ( PARTITION BY %s ORDER BY %s ASC ) AS rn FROM %s ) t WHERE rn = 1 ",
                columnInOrderStr, columnInOrderStr, distinctCol, distinctCol, this.frontTableName);
        String insertDistinct = String.format("INSERT INTO %s ( %s ) ( %s )", this.currentTableName, columnInOrderStr, distinctSelect);

        sqls.add(insertDistinct);

        updateSource();
        return sqls;
    }


    public void updateSource() {
        DamengTableQueueService flinkTableQueueService = SpringUtil.getBean(DamengTableQueueService.class);
        Map<String, List<DamengSourceColumn>> tableColumns = new ConcurrentHashMap<>(8);
        tableColumns.put(this.currentTableName, this.currentColumns);
        flinkTableQueueService.put(this.jobId, this.componentId, tableColumns);
    }
}
